package com.cbh.amadeus.client.receiver.tool.impl;

import com.cbh.amadeus.client.domain.MessageResultModel;
import com.cbh.amadeus.client.receiver.tool.MessageResultModelTool;
import com.cbh.amadeus.common.enums.MessageResult;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author ChenBiHui
 * @date 2025/4/30
 */
@Slf4j
public class DefaultMessageResultModelTool implements MessageResultModelTool {

    private final AtomicLong sequence = new AtomicLong(0);
    private final Map<String, MessageResultModel> record = new ConcurrentHashMap<>(64);

    @Override
    public String sequence() {
        return String.valueOf(sequence.incrementAndGet());
    }

    @Override
    public void retain(final MessageResultModel model) {
        if (model == null) {
            return;
        }
        String key = model.getTopic() + model.getMessageId() + model.getConsumer() + model.getLabel();
        MessageResultModel resultModel;
        synchronized (
                resultModel = record.computeIfAbsent(key, k -> model)
        ) {
            if (!model.equals(resultModel)) {
                resultModel.setInvokeResult(model.getInvokeResult());
                resultModel.setInvokeTime(model.getInvokeTime());
            }
        }
    }

    @Override
    public List<MessageResultModel> complete(String tenant, final String topic, final String messageId, final String name) {
        return record.values().stream()
                .filter(model -> tenant.equals(model.getTenant()))
                .filter(model -> topic.equals(model.getTopic()))
                .filter(model -> messageId.equals(model.getMessageId()))
                .filter(model -> name.equals(model.getConsumer()))
                .filter(model -> MessageResult.SUCCESS.name().equals(model.getInvokeResult()))
                .collect(Collectors.toList());
    }
}
